package com.example.toy.core.commit;

import com.example.toy.core.exception.NotExistsException;
import com.example.toy.core.utils.SchemaHelper;
import com.example.toy.core.version.ToyVersion;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.stream.Collectors;
import org.apache.avro.Schema;

/**
 * 提交记录视图
 *
 * @author shenbing
 * @date 2021/12/20
 */
public interface ToyCommitView {

  /**
   * 获取所有的提交记录
   *
   * @return
   */
  default List<CommitMsg> getCommits() {
    return getCommits(null, null, null, null, null, null);
  }

  /**
   * 获取指定类型的commits
   *
   * @param commitTypes 提交类型，如果为空，则匹配所有类型；不为空，当前提交历史和类型数组中任意一个匹配则返回
   * @param states 提交状态，如果为空，则匹配所有的状态；不为空，当前提交历史和状态数组中任意一个匹配则返回
   * @param from 起始版本，不为空，则从起始版本开始，包括起始版本；如果为空，则从第一个版本开始
   * @param to 结束版本，不为空，则从起始版本到结束版本，包括结束版本；如果为空，则到最新的版本。
   * @param fromTimestamp 开始时间
   * @param toTimestamp 结束时间
   * @return
   */
  List<CommitMsg> getCommits(
      CommitType[] commitTypes,
      CommitState[] states,
      ToyVersion from,
      ToyVersion to,
      Long fromTimestamp,
      Long toTimestamp);

  /**
   * 获取指定提交类型的最近的一次成功的提交历史
   *
   * @param commitType
   * @return
   */
  default Optional<CommitMsg> getLatestSucceedCommit(CommitType commitType) {
    List<CommitMsg> commits =
        getCommits(
            new CommitType[] {commitType},
            CommitState.SUCCEED_COMMIT_STATE,
            null,
            null,
            null,
            null);
    if (commits.isEmpty()) {
      return Optional.empty();
    } else {
      return Optional.of(commits.get(commits.size() - 1));
    }
  }

  /**
   * 获取所有未结束的提交历史
   *
   * @return
   */
  default List<CommitMsg> getUnCompletedCommits() {
    return getCommits(null, CommitState.UN_COMPLETED_COMMIT_STATE, null, null, null, null);
  }

  /**
   * 获取所有已经结束的提交历史
   *
   * @return
   */
  default List<CommitMsg> getCompletedCommits() {
    return getCommits(null, CommitState.COMPLETED_COMMIT_STATE, null, null, null, null);
  }

  /**
   * 获取指定版本的提交记录
   *
   * @param version
   * @return
   */
  default Optional<CommitMsg> getCommit(ToyVersion version) {
    List<CommitMsg> commits = getCommits(null, null, version, version, null, null);
    return commits.stream().findAny();
  }

  /**
   * 计算指定版本状态下的表的结构
   *
   * @param version
   * @return
   */
  default Schema getTableSchema(ToyVersion version) {
    List<CommitMsg> commits =
        getCommits(
            new CommitType[] {CommitType.INIT, CommitType.ALTER_COLUMN, CommitType.COMPACT},
            CommitState.SUCCEED_COMMIT_STATE,
            null,
            version,
            null,
            null);

    if (commits.isEmpty()) {
      throw new IllegalStateException("Can not find any commit from version: " + version);
    }

    CommitMsg commitWithBaseSchema = null;
    List<CommitMsg> alterColumns = new ArrayList<>();
    for (int i = commits.size() - 1; i >= 0; i--) {
      CommitMsg commit = commits.get(i);
      if (commit.getCommitType() == CommitType.ALTER_COLUMN) {
        alterColumns.add(0, commit);
      } else {
        commitWithBaseSchema = commit;
        break;
      }
    }
    if (commitWithBaseSchema == null) {
      throw new IllegalStateException(
          "Can not find any init or compact commit to resolve table schema");
    }
    Schema schema = commitWithBaseSchema.getSchema();
    for (CommitMsg alterColumn : alterColumns) {
      schema = SchemaHelper.mergeSchema(schema, alterColumn.getSchema());
    }
    return schema;
  }

  /**
   * 获取最近一个成功的提交记录
   *
   * @return
   * @throws NotExistsException 如果不存在成功的提交记录，则抛出异常
   */
  default CommitMsg getLatestSucceedCommit() throws NotExistsException {
    List<CommitMsg> commits =
        getCommits(null, CommitState.SUCCEED_COMMIT_STATE, null, null, null, null);
    if (commits.isEmpty()) {
      throw new NotExistsException("Can not find latest succeed version");
    }
    return commits.get(commits.size() - 1);
  }

  /**
   * 获取所有的可读数据的版本列表
   *
   * @return
   */
  default List<ToyVersion> getReadableVersions() {
    Optional<CommitMsg> compactCommit = getLatestSucceedCommit(CommitType.COMPACT);
    if (compactCommit.isPresent()) {
      return getCommits(
              null,
              CommitState.SUCCEED_COMMIT_STATE,
              compactCommit.get().getVersion(),
              null,
              null,
              null)
          .stream()
          .map(CommitMsg::getVersion)
          .collect(Collectors.toList());
    } else {
      return getCommits(null, CommitState.SUCCEED_COMMIT_STATE, null, null, null, null).stream()
          .map(CommitMsg::getVersion)
          .collect(Collectors.toList());
    }
  }
}
